package cn.codenest.logpublisher.util.service.impl;

import cn.codenest.logpublisher.util.service.LogService;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport;

/**
 * @author ：Hyman
 * @date ：Created in 2021/7/30 14:58
 * @description：
 * @modified By：
 * @version: $
 */
@Service
public class LogServiceImpl implements LogService {

    private ExecutorService executorService = Executors.newFixedThreadPool(10);

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Override
    public void sendMsg(String routeKey, String msg) {
        MessageProperties messageProperties = new MessageProperties();
        // 设置过期时间，单位：毫秒,30分钟
        messageProperties.setExpiration("1800000");
        messageProperties.setContentType("text/plain");
        messageProperties.setContentEncoding("UTF-8");
        byte[] msgBytes = msg.getBytes();
        Message message = new Message(msgBytes, messageProperties);
        CompletableFuture.runAsync(() -> rabbitTemplate.convertAndSend(
                "logTopicExchange",
                routeKey,
                message), executorService);
    }

    @Autowired
    ConnectionFactory connectionFactory;

    @Autowired
    SimpleMessageListenerContainer simpleMessageListenerContainer;

    @Override
    public void sendRPCMsg(String routeKey, String msg) {
        RabbitAdmin admin = new RabbitAdmin(connectionFactory);
        Queue replytoQueue = admin.declareQueue();
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setReplyTo(replytoQueue.getName());
        byte[] msgBytes = msg.getBytes();
        Message message = new Message(msgBytes, messageProperties);
        rabbitTemplate.convertAndSend(
                "eventTopicExchange",
                routeKey,
                message);
        Thread currentThread = Thread.currentThread();
        simpleMessageListenerContainer.addQueues(replytoQueue);
        simpleMessageListenerContainer.setMessageListener(new MessageListener() {

            @Override
            public void onMessage(Message message) {
                String str = new String(message.getBody());
                System.out.println(str);
                simpleMessageListenerContainer.removeQueues(replytoQueue);
                LockSupport.unpark(currentThread);
            }
        });
        LockSupport.park();
    }

}
